package saic.spark_stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka.KafkaUtils
import kafka.serializer.StringDecoder


/**
 * @author ZhiLi
 */
object FlumeTopic {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
  val ssc = new StreamingContext(conf, Seconds(1))
  
  val kafkaParams = Map[String, String](
//  "zk.connect" -> "10.129.96.12:2181,10.129.96.17:2181,10.129.96.12:2181",
      "zk.connect" -> "192.168.40.129:2181",
      "metadata.broker.list" -> "192.168.40.129:9092",
//  "key.deserializer" -> "kafka.serializer.StringEncoder",
//  "value.deserializer" -> "kafka.serializer.StringEncoder",
  "group.id" -> "1",
  "auto.offset.reset" -> "smallest",
  "enable.auto.commit" -> "true"    
  )
  
  val topics = Set[String]("mykafka")
  val readParallelism = 5
  println("KafkaTest Start!!!")
  val kafkaStream = KafkaUtils.createDirectStream[String, String,
    StringDecoder,StringDecoder](ssc, kafkaParams, topics)
  val flatMapStream = kafkaStream.flatMap(_._2)

  flatMapStream.print()
  
  ssc.start()
  ssc.awaitTermination()
  }
  
  
}